package mqtt;

import mqtt.handler.DefaultMqttCallback;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import util.PropertiesUtil;

public class MQTTv5Client {

	 private MqttAsyncClient mqttClient;
	    private final String brokerUrl;
	    private final String clientId;
	    private final String username;
	    private final String password;
	    private final MqttCallback callback;
	    private final MqttConnectionOptions connectionOptions;

	    public MQTTv5Client(String brokerUrl, String clientId, String username, String password, MqttCallback callback) {
	        this.brokerUrl = brokerUrl;
	        this.clientId = clientId;
	        this.username = username;
	        this.password = password;
	        this.callback = callback;
	        this.connectionOptions = createConnectionOptions();
	    }
	    
	    public MQTTv5Client(MqttCallback callback,String clientId) {
	        this.brokerUrl = PropertiesUtil.readByKey("mqtt.broker");
	        this.clientId = clientId;
	        this.username = PropertiesUtil.readByKey("mqtt.username");
	        this.password = PropertiesUtil.readByKey("mqtt.password");
	        this.callback = callback;
	        this.connectionOptions = createConnectionOptions();
	    }

	    private MqttConnectionOptions createConnectionOptions() {
	        MqttConnectionOptions options = new MqttConnectionOptions();
	        // 🎯 关键设置：确保只消费最新消息
	        options.setCleanStart(true);          // MQTT v5 中的 Clean Session
	        options.setSessionExpiryInterval(0L); // 会话立即过期，不保存任何状态
	        options.setAutomaticReconnect(true);  // 启用自动重连
	        options.setMaxReconnectDelay(30000);  // 最大重连间隔30秒
	        if (username != null) options.setUserName(username);
	        if (password != null) options.setPassword(password.getBytes());
	        
	        return options;
	    }

	    public void connect() throws MqttException {
	        if (mqttClient == null || !mqttClient.isConnected()) {
	            mqttClient = new MqttAsyncClient(brokerUrl, clientId, new MemoryPersistence());
	            mqttClient.setCallback(callback);
	            
	            IMqttToken connectToken = mqttClient.connect(connectionOptions);
	            connectToken.setActionCallback(new MqttActionListener() {
	                @Override
	                public void onSuccess(IMqttToken asyncActionToken) {
	                    System.out.println("Connected to broker: " + brokerUrl);
	                }

	                @Override
	                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
	                    System.err.println("Connection failed: " + exception.getMessage());
	                }
	            });
	            connectToken.waitForCompletion(); // 等待首次连接完成
	        }
	    }

	    public void disconnect() throws MqttException {
	        if (mqttClient != null && mqttClient.isConnected()) {
	            mqttClient.disconnect().waitForCompletion();
	        }
	    }

	    /**
	     * 发送字节 
	     * 
	     * 1. **QoS 0（最多一次）**：消息发送者只发送一次消息，不进行重试，接收方也不会确认。消息可能会丢失。
		   2. **QoS 1（至少一次）**：发送者确保消息至少被接收一次，但可能会重复。接收方需要确认，如果发送方没有收到确认，则会重发消息。
		   3. **QoS 2（恰好一次）**：通过两次握手确保消息只被传递一次。这是最高级别的服务质量，保证消息不会丢失也不会重复，但开销最大。
		   retained=true 时，Broker 会存储最后一条消息，新订阅者会立即收到该消息
	     * @param topic
	     * @param payload
	     * @param qos
	     * @param retained
	     * @throws MqttException
	     */
	    public void publish(String topic, byte[] payload, int qos, boolean retained) throws MqttException {
	        if (!mqttClient.isConnected()) {
	            connect(); // 确保连接存在
	        }
	        MqttMessage message = new MqttMessage(payload);
	        message.setQos(qos);
	        message.setRetained(retained);
	        mqttClient.publish(topic, message);
	    }
	    
	    /**
	     * 发送字符串
	     * @param topic
	     * @param msg
	     * @param qos
	     * @param retained
	     * @throws MqttException
	     */
	    public void publish(String topic, String msg, int qos, boolean retained) throws MqttException {
	        if (!mqttClient.isConnected()) {
	            connect(); // 确保连接存在
	        }
	        MqttMessage message = new MqttMessage(msg.getBytes());
	        message.setQos(qos);
	        message.setRetained(retained);
	        mqttClient.publish(topic, message);
	    }

	    public void subscribe(String topic, int qos) throws MqttException {
	        if (!mqttClient.isConnected()) {
	            connect();
	        }
	        mqttClient.subscribe(topic, qos);
	    }
	    
	    public void subscribe(String[] topic, int[] qos) throws MqttException {
	        if (!mqttClient.isConnected()) {
	            connect();
	        }
	        mqttClient.subscribe(topic, qos);
	    }

	    public static void main(String[] args) {
	        try {
	            // 示例用法
	        	 String username = "cytech";
	             String password = "Cytech@Mq_2100";
	            MQTTv5Client client = new MQTTv5Client(
	                    "tcp://172.18.66.150:1883",
	                    "java-client-" + System.currentTimeMillis(),
	                    username,
	                    password,
	                    new DefaultMqttCallback()
	            );

	            // 连接并订阅
	            client.connect();
	            client.subscribe("test", 1);

	            // 发布测试消息
	            //client.publish("test", "Hello MQTTv5 1".getBytes(), 0, false);

	            // 保持运行
	            //Thread.sleep(60000);
	            // 发布测试消息
	            System.out.println("发布第二条消息");
	            //client.publish("test", "Hello MQTTv5 2".getBytes(), 0, false);
	            //client.disconnect();
	        } catch (Exception e) {
	            e.printStackTrace();
	        }
	    }
	
}
